package com.rn.gypsophila.kafka.component;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.stereotype.Component;

/**
 * Description: 消息过滤器
 *
 * @author 然诺
 * @date 2021/02/02
 */
@Component
public class KafkaMsgFilter {
	@Autowired
	private ConsumerFactory consumerFactory;

	@Bean
	public ConcurrentKafkaListenerContainerFactory filterContainerFactory() {
		ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
		factory.setConsumerFactory(consumerFactory);
		// 被过滤的消息将被丢弃
		factory.setAckDiscarded(true);
		// 消息过滤策略
		factory.setRecordFilterStrategy(consumerRecord -> {
			return "KafkaMsgFilter_test".equals(consumerRecord.value());
			//返回true消息则被过滤
		});
		return factory;
	}


}
